package com.xfzcode.service.impl;

import com.xfzcode.pojo.CommonEntity;
import com.xfzcode.service.AnalysisService;
import com.xfzcode.thread.ResponseThreadLocal;
import com.xfzcode.utils.SearchTools;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.ParsedMultiBucketAggregation;
import org.elasticsearch.search.aggregations.bucket.ParsedSingleBucketAggregation;
import org.elasticsearch.search.aggregations.bucket.filter.Filters;
import org.elasticsearch.search.aggregations.bucket.filter.ParsedFilter;
import org.elasticsearch.search.aggregations.bucket.filter.ParsedFilters;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.aggregations.bucket.histogram.ParsedDateHistogram;
import org.elasticsearch.search.aggregations.bucket.histogram.ParsedHistogram;
import org.elasticsearch.search.aggregations.bucket.range.ParsedRange;
import org.elasticsearch.search.aggregations.bucket.range.Range;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.metrics.*;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;

/**
 * @Author: XMLee
 * @Date: 2022/8/4 18:26
 * @Description:
 */
@Service
public class AnalysisServiceImpl implements AnalysisService {

    private static final Logger logger = LoggerFactory.getLogger(AnalysisServiceImpl.class);

    @Autowired
    private RestHighLevelClient levelClient;


    /**
     * 指标聚合
     * @param commonEntity
     * @return
     * @throws Exception
     */
    @Override
    public Map<Object, Object> metricAgg(CommonEntity commonEntity) throws Exception {
        return responseResultConverter(commonEntity);
    }

    @Override
    public Map<Object, Object> bucketAgg(CommonEntity commonEntity) throws Exception {
        return responseResultConverter(commonEntity);
    }

    @Override
    public Map<String, Long> hotWords(CommonEntity commonEntity) throws Exception {
        //定义返回数据的对象
        Map<String, Long>  map=new LinkedHashMap<String, Long> ();
        //执行查询
        SearchResponse response=getSearchResponse(commonEntity);
        //获取数据
        Terms terms=response.getAggregations().get(response.getAggregations().getAsMap().entrySet().iterator().next().getKey());
        for(Terms.Bucket bucket:terms.getBuckets()){
            if(bucket.getKey()!=null){
                //key分组的字段
                String key=bucket.getKey().toString();
                //获取文档的个数
                Long  count=bucket.getDocCount();
                //将结果放到map里面
                map.put(key,count);
            }
        }
        return map;
    }

    /**
     * 结果转换器（桶聚合、指标聚合）
     * @param commonEntity
     * @return
     */
    private Map<Object, Object> responseResultConverter(CommonEntity commonEntity) throws Exception{
        //获取查询响应
        SearchResponse searchResponse = getSearchResponse(commonEntity);
        //定义返回数据
        Map<Object, Object> map = new HashMap<Object, Object>();
        //获取聚合数据
        Map<String, Aggregation> aggregationMap = searchResponse.getAggregations().asMap();
        //将数据放到线程中
        SearchTools.setResponseThreadLocal(searchResponse);
        for (Map.Entry<String, Aggregation> m : aggregationMap.entrySet()) {
            if (m.getValue() instanceof ParsedMultiBucketAggregation || m.getValue() instanceof ParsedSingleBucketAggregation) {
                //桶聚合
                bucketResultConverter(map, m);
            } else {
                //指标聚合
                metricResultConverter(map, m);
            }
        }
        mbCommonConverter(map);
        return map;
    }

    /**
     * 获取查询响应
     * @param commonEntity
     * @return
     * @throws Exception
     */
    private SearchResponse getSearchResponse(CommonEntity commonEntity) throws Exception {
        //定义查询请求
        SearchRequest searchRequest = new SearchRequest();
        //指定索引查询
        searchRequest.indices(commonEntity.getIndexName());
        //定义资源查询构建器
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //获取解析器
        XContentParser parser = SearchTools.getXContentParser(commonEntity);
        //将parse注册到构建器
        searchSourceBuilder.parseXContent(parser);
        //设置查询请求
        searchRequest.source(searchSourceBuilder);
        //执行查询
        SearchResponse searchResponse = levelClient.search(searchRequest, RequestOptions.DEFAULT);
        return searchResponse;
    }


    private void bucketResultConverter_back(Map<Object, Object> map, Map.Entry<String, Aggregation> m) {
        switch (m.getValue().getClass().getName()) {
            case "org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms":
                //获取桶数据
                List<? extends Terms.Bucket> list = ((ParsedStringTerms) m.getValue()).getBuckets();
                //定义排序map
                LinkedHashMap lmap = new LinkedHashMap();
                //循环桶里面的数据
                for (Terms.Bucket b : list) {
                    //处理terms数据
                    lmap.put(b.getKey(), b.getDocCount());
                }
                map.put("value", lmap);
                break;
            case "org.elasticsearch.search.aggregations.bucket.filter.ParsedFilter":
                //Filter聚合
                map.put("doc_count", ((ParsedFilter) m.getValue()).getDocCount());
                map.put("value", ((ParsedFilter) m.getValue()).getAggregations().asList());
                break;

            case "org.elasticsearch.search.aggregations.bucket.filter.ParsedFilters":
                //Filters聚合
                List<? extends Filters.Bucket> pfList = ((ParsedFilters) m.getValue()).getBuckets();
                for (Filters.Bucket b : pfList) {
                    map.put(b.getKey(), b.getDocCount());
                }
                break;
            case "org.elasticsearch.search.aggregations.bucket.histogram.ParsedHistogram":
                //histogram聚合
                List<? extends Histogram.Bucket> phList = ((ParsedHistogram) m.getValue()).getBuckets();
                for (Histogram.Bucket b : phList) {
                    map.put(b.getKey(), b.getDocCount());
                }
                break;
            //Range聚合
            case "org.elasticsearch.search.aggregations.bucket.range.ParsedRange":
                List<? extends Range.Bucket> prList = ((ParsedRange) m.getValue()).getBuckets();
                for (Range.Bucket b : prList) {
                    map.put(b.getKey(), b.getDocCount());
                }
                break;
            case "org.elasticsearch.search.aggregations.bucket.histogram.ParsedDateHistogram":
                //Date聚合
                List<? extends Histogram.Bucket> pdList = ((ParsedDateHistogram) m.getValue()).getBuckets();
                for (Histogram.Bucket b : pdList) {
                    map.put(b.getKey(), b.getDocCount());
                }
                break;
            default:
                break;
        }
    }
    /**
     * 公共数据处理
     * @param map
     */
    private void mbCommonConverter(Map<Object, Object> map) {
        if (!CollectionUtils.isEmpty(ResponseThreadLocal.get())) {
            //从线程中取出数据
            map.put("list", ResponseThreadLocal.get());
            //清空本地线程数据，否则将内存泄露
            ResponseThreadLocal.clear();
        }
    }

    /**
     * 指标聚合
     * @param map
     * @param m
     */
    private void metricResultConverter_back(Map<Object, Object> map, Map.Entry<String, Aggregation> m) {
        switch (m.getValue().getClass().getName()) {
            //平均值
            case "org.elasticsearch.search.aggregations.metrics.ParsedAvg":
                map.put("value", ((ParsedAvg) m.getValue()).getValue());
                break;
            //最大值
            case "org.elasticsearch.search.aggregations.metrics.ParsedMax":
                map.put("value", ((ParsedMax) m.getValue()).getValue());
                break;
            //最小值
            case "org.elasticsearch.search.aggregations.metrics.ParsedMin":
                map.put("value", ((ParsedMin) m.getValue()).getValue());
                break;
            //求和
            case "org.elasticsearch.search.aggregations.metrics.ParsedSum":
                map.put("value", ((ParsedSum) m.getValue()).getValue());
                break;
            //不重复的值
            case "org.elasticsearch.search.aggregations.metrics.ParsedCardinality":
                map.put("value", ((ParsedCardinality) m.getValue()).getValue());
                break;
            //扩展统计
            case "org.elasticsearch.search.aggregations.metrics.ParsedExtendedStats":
                map.put("count", ((ParsedExtendedStats) m.getValue()).getCount());
                map.put("min", ((ParsedExtendedStats) m.getValue()).getMin());
                map.put("max", ((ParsedExtendedStats) m.getValue()).getMax());
                map.put("avg", ((ParsedExtendedStats) m.getValue()).getAvg());
                map.put("sum", ((ParsedExtendedStats) m.getValue()).getSum());
                map.put("sum_of_squares", ((ParsedExtendedStats) m.getValue()).getSumOfSquares());
                map.put("variance", ((ParsedExtendedStats) m.getValue()).getVariance());
                map.put("std_deviation", ((ParsedExtendedStats) m.getValue()).getStdDeviation());
                map.put("upper", ((ParsedExtendedStats) m.getValue()).getStdDeviationBound(ExtendedStats.Bounds.UPPER));
                map.put("lower", ((ParsedExtendedStats) m.getValue()).getStdDeviationBound(ExtendedStats.Bounds.LOWER));
                break;
            //状态统计
            case "org.elasticsearch.search.aggregations.metrics.ParsedStats":
                map.put("count", ((ParsedStats) m.getValue()).getCount());
                map.put("min", ((ParsedStats) m.getValue()).getMin());
                map.put("max", ((ParsedStats) m.getValue()).getMax());
                map.put("avg", ((ParsedStats) m.getValue()).getAvg());
                map.put("sum", ((ParsedStats) m.getValue()).getSum());
                break;
            //百分位度量
            case "org.elasticsearch.search.aggregations.metrics.ParsedTDigestPercentiles":
                for (Iterator<Percentile> iterator = ((ParsedTDigestPercentiles) m.getValue()).iterator(); iterator.hasNext(); ) {
                    Percentile p = iterator.next();
                    map.put(p.getPercent(), p.getValue());
                }
                break;
            //百分位等级
            case "org.elasticsearch.search.aggregations.metrics.ParsedTDigestPercentileRanks":
                for (Iterator<Percentile> iterator = ((ParsedTDigestPercentileRanks) m.getValue()).iterator(); iterator.hasNext(); ) {
                    Percentile p = iterator.next();
                    map.put(p.getValue(), p.getPercent());
                }
                break;
            default:
                break;
        }
    }

    /**
     * 桶聚合结果转换器
     * @param map
     * @param m
     */
    private void bucketResultConverter(Map<Object, Object> map, Map.Entry<String, Aggregation> m) {
        if (m.getValue() instanceof ParsedStringTerms) {
            //获取桶数据
            List<? extends Terms.Bucket> list = ((ParsedStringTerms) m.getValue()).getBuckets();
            //定义排序map
            LinkedHashMap lmap = new LinkedHashMap();
            //循环桶里面的数据
            for (Terms.Bucket b : list) {
                //处理terms数据
                lmap.put(b.getKey(), b.getDocCount());
            }
            map.put("value", lmap);
        }
        //Filter聚合
        if (m.getValue() instanceof ParsedFilter) {
            map.put("doc_count", ((ParsedFilter) m.getValue()).getDocCount());
            map.put("value", ((ParsedFilter) m.getValue()).getAggregations().asList());
        }
        //Filters聚合
        if (m.getValue() instanceof ParsedFilters) {
            List<? extends Filters.Bucket> list = ((ParsedFilters) m.getValue()).getBuckets();
            for (Filters.Bucket b : list) {
                map.put(b.getKey(), b.getDocCount());
            }
        }
        //histogram聚合
        if (m.getValue() instanceof ParsedHistogram) {
            List<? extends Histogram.Bucket> list = ((ParsedHistogram) m.getValue()).getBuckets();
            for (Histogram.Bucket b : list) {
                map.put(b.getKey(), b.getDocCount());
            }
        }
        //Range聚合
        if (m.getValue() instanceof ParsedRange) {
            List<? extends Range.Bucket> list = ((ParsedRange) m.getValue()).getBuckets();
            for (Range.Bucket b : list) {
                map.put(b.getKey(), b.getDocCount());
            }
        }
        //Date聚合
        if (m.getValue() instanceof ParsedDateHistogram) {
            List<? extends Histogram.Bucket> list = ((ParsedDateHistogram) m.getValue()).getBuckets();
            for (Histogram.Bucket b : list) {
                map.put(b.getKey(), b.getDocCount());
            }
        }
    }

    /**
     * 指标聚合结果转换器
     * @param map
     * @param m
     */
    private void metricResultConverter(Map<Object, Object> map, Map.Entry<String, Aggregation> m) {
        //平均值
        if (m.getValue() instanceof ParsedAvg) {
            map.put("value", ((ParsedAvg) m.getValue()).getValue());
        }
        //最大值
        if (m.getValue() instanceof ParsedMax) {
            map.put("value", ((ParsedMax) m.getValue()).getValue());
        }
        //最小值
        if (m.getValue() instanceof ParsedMin) {
            map.put("value", ((ParsedMin) m.getValue()).getValue());
        }
        //求和
        if (m.getValue() instanceof ParsedSum) {
            map.put("value", ((ParsedSum) m.getValue()).getValue());
        }

        //求唯一值
        if (m.getValue() instanceof ParsedCardinality) {
            map.put("value", ((ParsedCardinality) m.getValue()).getValue());
        }
        //状态统计
        if (m.getValue() instanceof ParsedStats) {
            map.put("count", ((ParsedStats) m.getValue()).getCount());
            map.put("min", ((ParsedStats) m.getValue()).getMin());
            map.put("max", ((ParsedStats) m.getValue()).getMax());
            map.put("avg", ((ParsedStats) m.getValue()).getAvg());
            map.put("sum", ((ParsedStats) m.getValue()).getSum());
        }
        //扩展统计
        if (m.getValue() instanceof ParsedExtendedStats) {
            map.put("count", ((ParsedExtendedStats) m.getValue()).getCount());
            map.put("min", ((ParsedExtendedStats) m.getValue()).getMin());
            map.put("max", ((ParsedExtendedStats) m.getValue()).getMax());
            map.put("avg", ((ParsedExtendedStats) m.getValue()).getAvg());
            map.put("sum", ((ParsedExtendedStats) m.getValue()).getSum());

            map.put("sum_of_squares", ((ParsedExtendedStats) m.getValue()).getSumOfSquares());
            map.put("variance", ((ParsedExtendedStats) m.getValue()).getVariance());
            map.put("std_deviation", ((ParsedExtendedStats) m.getValue()).getStdDeviation());
            map.put("upper", ((ParsedExtendedStats) m.getValue()).getStdDeviationBound(ExtendedStats.Bounds.UPPER));
            map.put("lower", ((ParsedExtendedStats) m.getValue()).getStdDeviationBound(ExtendedStats.Bounds.LOWER));
        }
        //百分位度量
        if (m.getValue() instanceof ParsedTDigestPercentiles) {
            for (Iterator<Percentile> iterator = ((ParsedTDigestPercentiles) m.getValue()).iterator(); iterator.hasNext(); ) {
                Percentile p = (Percentile) iterator.next();
                map.put(p.getPercent(), p.getValue());
            }
        }
        //百分位等级
        if (m.getValue() instanceof ParsedTDigestPercentileRanks) {
            for (Iterator<Percentile> iterator = ((ParsedTDigestPercentileRanks) m.getValue()).iterator(); iterator.hasNext(); ) {
                Percentile p = (Percentile) iterator.next();
                map.put(p.getValue(), p.getPercent());
            }
        }
    }
}
